package com.zimmor.mq.rocket_train.base;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.Collection;
import java.util.List;

/**
 * @Author:zimmor
 * @Date: 2023/2/15 9:35
 * @Description pull的方式消费的assign的模式
 * @version: 1.0.0
 */

public class PullAssConsumer {
    public static void main(String[] args) throws MQClientException {
        //创建消费者
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("consumer-group6");
        //设置namesrv
        consumer.setNamesrvAddr("192.168.204.136:9876;192.168.204.137:9876");
        //设置手痛提交
        consumer.setAutoCommit(false);
        //启动消费者
        consumer.start();
        //获取要分配的队列集合
        Collection<MessageQueue> list = consumer.fetchMessageQueues("base1");
        //设置要分配的队列
        consumer.assign(list);
        //监听队列
        boolean running = true;
        try{
            while (running) {
                List<MessageExt> poll = consumer.poll();
                for (MessageExt messageExt : poll) {
                    System.out.println("msg:" + new String(messageExt.getBody()));
                }
                consumer.commitSync();
                System.out.println("========================");
            }
        }finally {
                consumer.shutdown();
        }
        //关闭消费者
//        consumer.shutdown();
    }
}
